package com.batchengine.manager;

import com.batchengine.pool.WorkerPool;
import static com.batchengine.pool.WorkerPoolFactory.createExecutorPool;
import static com.batchengine.pool.WorkerPoolFactory.createProducerConsumerPool;
import com.batchengine.task.RecurringJob;
import static com.batchengine.task.TaskUtil.getRecurringJobMap;
import com.batchengine.test.HelloWorkSelector;
import com.batchengine.test.HelloWorldDataGatherer;
import com.batchengine.util.EnvUtil;
import static com.batchengine.util.LoggingFactory.createLogger;
import com.batchengine.worker.ConsumerFactory;
import com.batchengine.worker.DataGatherer;
import com.batchengine.worker.PCStopSignal;
import com.batchengine.worker.WorkSelector;
import org.apache.log4j.Logger;

import java.util.Map;

/**
 * Created by sunil.incubator@yahoo.com.
 * User: sunil.kalva
 * Date: Jul 12, 2009
 */
public class UserWorkerManager extends WorkerManager {

    private static final Logger logger = createLogger();

    public UserWorkerManager(String name) {
        super(name);
    }

    public synchronized void init() {

        logger.debug("calling user UserWorkerManager");
        Map<Long, RecurringJob> map = getRecurringJobMap();
        for (Long o : map.keySet()) {

            if (100 < o) {
                continue;
            }
            RecurringJob job = map.get(o);
            if (null == job.getTaskInfo().getSelctorClazz() || 0 == job.getTaskInfo().getSelctorClazz().trim().length()) {
                createExecutorRecurringPool(job);
                continue;
            }
            createPCRecurringPool(job);
        }
    }

    protected void createExecutorRecurringPool(RecurringJob job) {

        WorkerPool pool = createExecutorPool(job.getMinWorkers(), job.getMaxWorkers(), job.getTaskInfo().getTaskId());
        pool.setDelaySecs(job.getDelay());
        pool.setName(job.getInstanceName());
        addWorkerPool(pool);
//        pool.startWorkers();
    }

    private void createPCRecurringPool(RecurringJob job) {

        WorkSelector workSelector = (WorkSelector) EnvUtil.createObjectz(job.getTaskInfo().getSelctorClazz());
        DataGatherer dataGatherer = new DataGatherer(String.format("%s-Gatherer", job.getInstanceName()), job.getJobId(), workSelector);
        WorkerPool pool = createProducerConsumerPool(job.getMinWorkers(), job.getMaxWorkers(), job.getTaskInfo().getTaskId(), dataGatherer);
        pool.setDelaySecs(job.getDelay());
        if (null != job.getTaskInfo().getSignalClazz() && 0 < job.getTaskInfo().getSignalClazz().trim().length() && !"null".equalsIgnoreCase(job.getTaskInfo().getSignalClazz().trim())) {
            PCStopSignal signalPC = (PCStopSignal) EnvUtil.createObjectz(job.getTaskInfo().getSignalClazz());
            pool.setSignal(signalPC);
        }
        pool.setName(job.getInstanceName());
        addWorkerPool(pool);
//        pool.startWorkers();
    }

    private void createTestPool() {

        HelloWorkSelector workSelector = new HelloWorkSelector();
        DataGatherer dataGatherer = new HelloWorldDataGatherer("HW-DataGatherer", 1, workSelector);
        WorkerPool producerConsumerPool = createProducerConsumerPool(2, 10, 2l, new ConsumerFactory(workSelector.getWorkQ()), dataGatherer);
        addWorkerPool(producerConsumerPool);
        producerConsumerPool.setName("HelloWorld");
        producerConsumerPool.startWorkers();
    }
}
